from datetime import timedelta, datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago


default_args = {
    'owner': 'airflow',
    # 必须要有start date
    # schedule_interval定义在下面的dag中，默认是一天。
    # 这里的逻辑是：
    #  1. start date是第一次execution date。 如果start_date + interval已经pass了，那么就会启动dag，start date就是第一次的execution time。
    #  2. 如果在ui上点了On开启，当之前最新一次的execution date + interval * 2 <= 当前时间的时候，执行下一次execution。
    #     execution date设置为之前最新一次的 + interval
    #  3. execution会依次进行，就算之前的一次失败了也是如此
    #  例如，这里设置是7：30为start date。那么当8: 35导入的时候，如果打开On，那么会自动依次跑 7:30，7:40, 7:50，8:00, 8:10，8:20 
    #  这里的时间都是指的execution date。
    #  当8: 40过了后，会执行exection date为8:30的。
    # 只算schedule自动调度的，手动跑的不影响这个调度过程。
    # 虽然可以给task设定不同的start date, 但还是推荐在default args里面设置统一的start date，这样便于理解。
    # start_date 默认是按 utc 时间算的
    'start_date': datetime(2021, 7, 5, 6, 30),
}

dag = DAG(
    'test-time-normal',
    schedule_interval=timedelta(minutes=10),
    default_args=default_args,
    description='test',
    tags=['test'],
)

t1 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag,
)

t2 = BashOperator(
    task_id='print_env',
    bash_command='printenv',
    dag=dag,
)

sleep = BashOperator(
    task_id='sleep',
    bash_command='sleep 30s',
    dag=dag,
)

sleep >> t1 >> t2